package handler

import (
	"context"
	"github.com/Shopify/sarama"
	"github.com/cenkalti/backoff/v4"
	"github.com/olivere/elastic/v7"
	"go.uber.org/zap"
	"kee/dao"
	"kee/utils/zaplog"
	"log"
	"os"
	"os/signal"
	"sync"
	"syscall"
	"time"
)

// 存储数据
func (impl *ImplKee) StorageData() {

	ctx, cancel := context.WithCancel(context.Background())
	/**
	 * Setup a new Sarama consumer group
	 */
	consumer := Consumer{
		ready:       make(chan bool),
		esClient:    impl.Dao.ESClient,
		esIndexName: impl.Config.ESConfig.EsIndex,
	}

	wg := &sync.WaitGroup{}
	wg.Add(1)
	go func() {
		defer wg.Done()
		for {
			// `Consume` should be called inside an infinite loop, when a
			// server-side rebalance happens, the consumer session will need to be
			// recreated to get the new claims
			if err := impl.Dao.ConsumerGroupClient.ConsumerGroupClient.Consume(ctx, impl.Config.KafkaClientConfig.Topics, &consumer); err != nil {
				log.Panicf("Error from consumer: %v", err)
			}
			// check if context was cancelled, signaling that the consumer should stop
			if ctx.Err() != nil {
				return
			}
			consumer.ready = make(chan bool)
		}
	}()

	<-consumer.ready // Await till the consumer has been set up
	log.Println("Sarama consumer up and running!...")

	sigterm := make(chan os.Signal, 1)
	signal.Notify(sigterm, syscall.SIGINT, syscall.SIGTERM)
	select {
	case <-ctx.Done():
		log.Println("terminating: context cancelled")
	case <-sigterm:
		log.Println("terminating: via signal")
	}
	cancel()
	wg.Wait()
	if err := impl.Dao.ConsumerGroupClient.ConsumerGroupClient.Close(); err != nil {
		log.Panicf("Error closing client: %v", err)
	}
}

// Consumer represents a Sarama consumer group consumer
type Consumer struct {
	ready       chan bool
	esClient    *elastic.Client
	esIndexName string
}

// Setup is run at the beginning of a new session, before ConsumeClaim
func (consumer *Consumer) Setup(sarama.ConsumerGroupSession) error {
	// Mark the consumer as ready
	close(consumer.ready)
	return nil
}

// Cleanup is run at the end of a session, once all ConsumeClaim goroutines have exited
func (consumer *Consumer) Cleanup(sarama.ConsumerGroupSession) error {
	return nil
}

// ConsumeClaim must start a consumer loop of ConsumerGroupClaim's Messages().
func (consumer *Consumer) ConsumeClaim(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error {
	// NOTE:
	// Do not move the code below to a goroutine.
	// The `ConsumeClaim` itself is called within a goroutine, see:
	// https://github.com/Shopify/sarama/blob/master/consumer_group.go#L27-L29
	for message := range claim.Messages() {

		data := string(message.Value)
		err := dao.AddStockData(context.Background(), consumer.esClient, consumer.esIndexName, &data)

		if err != nil {
			// 缓存到达限制大小，无法往里插入数据
			// 指数退避算法会利用抖动（随机延迟）来防止连续的冲突。 效果如下，每次间隔的时间都是指数上升，另外加了少许的随机。
			b := backoff.NewExponentialBackOff()
			b.MaxElapsedTime = 10 * time.Second
			err := backoff.Retry(func() error {
				err := dao.AddStockData(context.Background(), consumer.esClient, consumer.esIndexName, &data)
				time.Sleep(1 * time.Second)
				if err != nil {
					zaplog.Trace("es").Error("ConsumeClaim",
						zap.String("data", data),
						zap.String("topic", message.Topic),
						zap.Time("timestamp", message.Timestamp),
						zap.Error(err),
					)
				}
				return err
			}, b)

			if err != nil {
				zaplog.Trace("es").Error("backOff", zap.Error(err))
			}
		}

		session.MarkMessage(message, "")
	}

	return nil
}
